/**********************************************************************
 * FILE : GlobalCmd.java
 * CREATE DATE : 2011-11-25
 * DESCRIPTION :
 *		
 *      
 * CHANGE HISTORY LOG
 *---------------------------------------------------------------------
 * NO.|    DATE    |     NAME     |     REASON     | DESCRIPTION
 *---------------------------------------------------------------------
 * 1  | 2011-11-25 |  Sting  |    创建草稿版本
 *---------------------------------------------------------------------              
 **********************************************************************
 */
package com.sgm.de.benchmark.act;

import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;

import com.sgm.de.benchmark.BKCode;
import com.sgm.de.benchmark.JMSProcesser;
import com.sgm.de.benchmark.MessageGen;
import com.sgm.de.benchmark.ProcessContext;
import com.sgm.de.benchmark.Worker;
import com.sgm.dms.de.driver.IMessage;

/**
 * @author Sting<p>
 * TODO 这里描述该类的主要作用，以及使用时需要注意的事项
 */
public class GlobalAct extends AbstractAct {

    private String cmdName;
    private final Object[] args;
    private ArrayBlockingQueue<IMessage> queue;
    private ThreadPoolExecutor executor;

    public GlobalAct(String actor, Object... args) {
        this.cmdName = actor;
        this.args = args;
        messagearray = initArray();
    }


    @Override
    public void exec(ProcessContext ctx) throws Exception {
        JMSProcesser processor = ctx.processor;
        PrintWriter toAgent = ctx.response;
        if (cmdName.equalsIgnoreCase("hello")) {
            System.out.println("Hello from the server!");
            ctx.response.println("ack");
            ctx.response.flush();
        }
        else if (cmdName.equalsIgnoreCase("time")) {
            System.out.println("set time mode");
            processor.testType = BKCode.TEST_TIME;
            processor.timeSecLimit = Integer.parseInt(args[0].toString());
            toAgent.println("ack");
            toAgent.flush();
        } else if (cmdName.equalsIgnoreCase("addThread")) {
            int addNum = Integer.parseInt(args[0].toString());
            ctx.threadCount += addNum;

        } else if (cmdName.equalsIgnoreCase("runTest")) {
            System.out.println("Process-" + Thread.currentThread().getId() + ": Mission Start!!");
            Worker.toBeStop = false;
            Semaphore semaphore = new Semaphore(0);
            executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(ctx.threadCount);
            queue = new ArrayBlockingQueue<IMessage>(50000);
            if (ctx.config.getQueueName() != null && ctx.queuePool == null) {
                String[] queues = ctx.config.getQueueName().split(";");// 分号分割线程
                ArrayBlockingQueue<String[]> queuepool = new ArrayBlockingQueue<String[]>(5000);
                for (int i = 0; i < ctx.threadCount; i++) {// 每个线程陪一份poolqueue,线程从pool中取名称会删除
                    for(String q:queues){
                        String[] threadQueue = q.split(",");// 逗号线程内使用多个queue
                        queuepool.add(threadQueue);
                    }
                }
                ctx.queuePool = queuepool;
            } else if (ctx.config.getQueueName() == null && ctx.queuePool == null) {
                System.err.println("error config: no queue name !!!");
                return;
            }
            for (int i = 0; i < ctx.threadCount; i++) {
                Worker onework = new Worker(semaphore, queue, ctx);
                executor.execute(onework);
            }
            try {
                Thread.currentThread().sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            // 数量测试
            if (BKCode.TEST_QUANTITY.equals(processor.testType)) {
                semaphore.release(ctx.threadCount);
                int lefted = processor.messageLimit;
                while (lefted-- > 0) {
                    Thread.sleep(50);
                    if (lefted % 100 == 0) {
                        System.out.println("remain message quantity:" + lefted);
                    }
                    IMessage msg = getMessage(ctx.config.getMessageSize(), ctx.config.getMessageMixed());
                    queue.put(msg);
                }
                Worker.toBeStop = true;
                while (executor.getActiveCount() > 0) {
                    Thread.sleep(1000);
                }
                executor.shutdown();
                System.out.println("Process-" + Thread.currentThread().getId() + ": Mission Complete!!");
            } else if (BKCode.TEST_TIME.equals(processor.testType)) {// 时间测试
                semaphore.release(ctx.threadCount);
                Calendar cal = Calendar.getInstance();
                cal.add(Calendar.SECOND, processor.timeSecLimit);
                System.out.printf("until to %tT\n", cal);
                long until = System.currentTimeMillis() + processor.timeSecLimit * 1000;
                while (System.currentTimeMillis() < until) {
                    if ((until - System.currentTimeMillis()) / 1000 > 3 && queue.size() < 500) {
                        int count = 500;
                        while (count-- > 0) {
                            IMessage msg = getMessage(ctx.config.getMessageSize(), ctx.config.getMessageMixed());
                            queue.put(msg);
                        }
                        System.out.println("remain message quantity:" + queue.size());
                    } else {
                        IMessage msg = getMessage(ctx.config.getMessageSize(), ctx.config.getMessageMixed());
                        queue.put(msg);
                    }
                }

                queue.drainTo(new ArrayList<IMessage>());
                Worker.toBeStop = true;
                while (executor.getActiveCount() > 0) {
                    Thread.sleep(1000);
                }
                executor.shutdown();
                System.out.println("Process-" + Thread.currentThread().getId() + ": Mission Complete!!");
            } else {
                semaphore.release(ctx.threadCount);
                System.out.println("Process-" + Thread.currentThread().getId() + ": Mission cycle!!");
            }

        } else if (cmdName.equalsIgnoreCase("stop")) {
            processor.distory();
        }
    }

    /**
     * @return
     * @throws InstantiationException
     * @throws IllegalAccessException
     */
    private List<IMessage> messageList = new ArrayList<IMessage>();
    Random r = new Random();
    public IMessage getMessage(int size, int mixSize) throws InstantiationException, IllegalAccessException {
        if (messageList.size() == 0) {
            int total = mixSize > 1 ? mixSize : 1;
            for (int i = 0; i < total; i++) {
                IMessage newMessage = MessageGen.newMessage();
                newMessage.setMessageType(IMessage.MessageType.Test);

                int onesize = total == 1 ? size * 10 : r.nextInt(size * 10);
                for (int j = 0; j < onesize; j++) {
                    newMessage.write(messagearray[r.nextInt(10)]);
                }
                messageList.add(newMessage);
            }
        }

        return mixSize > 1 ? messageList.get(r.nextInt(messageList.size())) : messageList.get(0);
    }

    @Override
    public String getCmdName() {
        return cmdName;
    }

    private byte[][] messagearray = null;

    private static byte[][] initArray() {
        int size = 1024;
        Random r = new Random();
        byte[][] array = new byte[10][size];
        byte[] row = new byte[size];
        for (int i = 0; i < 10; i++) {
            for (int j = 0; j < size; j++) {
                row[j] = (byte) r.nextInt();
            }
            array[i] = row;
        }
        return array;
    }

}
